package com.atguigu.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @Author lzc
 * @Date 2023/6/1 14:09
 */
public class HbaseUtil {
    public static AsyncConnection getHbaseAsyncConnection() {
        Configuration conf = HBaseConfiguration.create();
        
        // 配置 Zookeeper
        conf.set("hbase.zookeeper.quorum", "hadoop162");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        
        try {
            return ConnectionFactory.createAsyncConnection(conf).get();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
    
/**
 * 异步的方式从 HBase 读取维度数据
 * @param conn 到 HBase 的异步连接
 * @param nameSpace 命名空间
 * @param tableName 表名
 * @param rowKey rowKey
 * @return 读到的维度数据, 如果返回 null,表示没有读到维度数据
 */
public static JSONObject asyncReadDim(AsyncConnection conn, String nameSpace, String tableName, String rowKey) {
    
    AsyncTable<AdvancedScanResultConsumer> asyncTable = conn.getTable(TableName.valueOf(nameSpace, tableName));
    Get get = new Get(Bytes.toBytes(rowKey));
    CompletableFuture<Result> future = asyncTable.get(get);
    try {
        Result result = future.get();
        JSONObject dim = new JSONObject();
        for (Cell cell : result.listCells()) {
            String name = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            dim.put(name, value);
        }
        return dim;
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
    
}
    
    public static void main(String[] args) throws ExecutionException, InterruptedException {
    
    
    }
    
    public static JSONObject readDim(Connection conn,
                                     String nameSpace,
                                     String tableName,
                                     String rowKey) {
        
        try(Table table = conn.getTable(TableName.valueOf(nameSpace, tableName))) {
            
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            List<Cell> cells = result.listCells();  // 得到所有的列
            JSONObject obj = new JSONObject();
            
            for (Cell cell : cells) {
                String name = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                obj.put(name, value);
            }
            return obj;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    
    
    public static void closeHbaseConnection(Connection conn) {
        if (conn != null && !conn.isClosed()) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    public static void dropTable(Connection hbaseConn, String nameSpace, String tableName) {
        try (Admin admin = hbaseConn.getAdmin()) {
            if (!admin.tableExists(TableName.valueOf(nameSpace, tableName))) {
                System.out.println("要删除的表: " + nameSpace + ":" + tableName + " 不存在...");
                return;
            }
            admin.disableTable(TableName.valueOf(nameSpace, tableName));
            admin.deleteTable(TableName.valueOf(nameSpace, tableName));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    /**
     * 获取 hbase 连接
     *
     * @return
     */
    public static Connection getHbaseConnection() {
        Configuration conf = HBaseConfiguration.create();
        
        // 配置 Zookeeper
        conf.set("hbase.zookeeper.quorum", "hadoop162");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        try {
            return ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    
    
    /**
     * 在 hbase 中建表
     *
     * @param hbaseConn 到 hbase 的链接对象
     * @param nameSpace 命名空间
     * @param tableName 表名
     * @param families  列族
     */
    public static void createHbaseTable(Connection hbaseConn,
                                        String nameSpace,
                                        String tableName,
                                        String... families) {
        if (families.length < 1) {
            throw new IllegalArgumentException("请至少需要一个列族名...");
        }
        
        try (Admin admin = hbaseConn.getAdmin()) {
            //判断表是否存在
            if (admin.tableExists(TableName.valueOf(nameSpace, tableName))) {
                System.out.println(nameSpace + ":" + tableName + "已存在");
                return;
            }
            TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(TableName.valueOf(nameSpace, tableName));
            
            //指定列族
            for (String family : families) {
                ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder
                    .newBuilder(Bytes.toBytes(family)).build();
                builder.setColumnFamily(familyDescriptor);
            }
            admin.createTable(builder.build());
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public static void putRow(Connection conn,
                              String nameSpace,
                              String tableName,
                              String rowKey,
                              String family,
                              String[] columns,
                              String[] values) throws IOException {
        // 获取 table 对象
        Table table = conn.getTable(TableName.valueOf(nameSpace, tableName));
        // 根据传入的参数创建 put 对象
        Put put = new Put(Bytes.toBytes(rowKey));
        for (int i = 0; i < columns.length; i++) {
            String column = columns[i];
            String value = values[i];
            if (value != null) { // 有部分列的值是 null, 不用写出
                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
            }
        }
        table.put(put);
        table.close();
    }
    
    public static void delRow(Connection hbaseConn,
                              String nameSpace,
                              String tableName,
                              String rowKey) throws IOException {
        Table table = hbaseConn.getTable(TableName.valueOf(nameSpace, tableName));
        Delete del = new Delete(Bytes.toBytes(rowKey));
        table.delete(del);
        table.close();
    }
    
    public static String getBaseDicLookUpDDL() {
        return "CREATE TABLE base_dic (" +
            // 原子类型会自动识别为 hbase 的 RowKey. 列名随意
            " dic_code string," +
            // 声明列族,类型必须是 Row 类型, 嵌套的字段名是 hbase 中的列.
            " info ROW<dic_name string>," +
            " PRIMARY KEY (dic_code) NOT ENFORCED" +
            ")with(" +
            " 'connector' = 'hbase-2.2'," +
            " 'zookeeper.quorum' = 'hadoop162,hadoop163,hadoop164', " +
            " 'table-name' = 'gmall:dim_base_dic', " +
            " 'lookup.async' = 'true', " +
            " 'lookup.cache' = 'PARTIAL'," +
            " 'lookup.partial-cache.max-rows' = '200'," +
            //                            " 'lookup.partial-cache.expire-after-write' = '20 second', " +
            " 'lookup.partial-cache.expire-after-access' = '20 second'" +
            ")";
    }
}
